package com.aixin.project.util;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.concurrent.atomic.AtomicInteger;

import com.aixin.project.conf.Configure;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.log4j.Logger;

import com.danga.MemCached.MemCachedClient;
import com.danga.MemCached.SockIOPool;

/**
 * Created by IntelliJ IDEA.
 * User: Sufeng
 * Date: 12-4-14
 * Time: 下午4:23
 * To change this template use File | Settings | File Templates.
 */
public class AxMemcached {
	 private static final Logger  logger = Logger.getLogger(AxMemcached.class);
	    public static AtomicInteger THREAD_COUNT = new AtomicInteger(0);

		// create a static client as most installs only need
		// a single instance
		protected static MemCachedClient mcc = new MemCachedClient();
		protected static boolean isMCInitSuccess = true;

		// this just is a dummy implementation using HashMap before hooking up with Memcached.
		private static HashMap cache = new HashMap();
		private static Timer axCacheTimer = new Timer(true);
		private static HashMap timerTasks = new HashMap();

		// set up connection pool once at class load
		static {

			init();
			mcc.setPoolName("AXMEMCACHED");
			// kick off the memcached server monitoring background thread

			MemcachedTimerTask mctt = new MemcachedTimerTask(AxMemcached.class.getName());
			int mcdelay = Configure.MEMCACHED_STATUS_CHECK_INTERVAL;
			long delay = mcdelay * 1000;
            axCacheTimer.schedule(mctt, delay, delay);

		}

		public static void purgeCache() {
			if (!cache.isEmpty())cache.clear();
		}

		public static void expireCache(String key) {
			cache.remove(key);
		}

		public static boolean check() {
			String key = "ax";
			String value = "ax";
			mcc.set(key, value);
			String vu = (String)mcc.get(key);
			if (value.equals(vu)) {
				//isMCInitSuccess = true;
				//logger.error("Initialized check true: "+vu);
				return true;
			}
			else {
				logger.error("@@@@ Initialized check false: "+vu);
				//isMCInitSuccess = false;
				return false;
			}
		}

		public static long getCounter(String counterKey) {
			logger.error("miscmemcached get counterkey: "+counterKey);
			long count = -1;
			try {
	        	THREAD_COUNT.incrementAndGet();
				count = mcc.getCounter(counterKey);
			} catch (Exception e) {
				// NOOP
			} finally {
	            THREAD_COUNT.decrementAndGet();
			}
			return count;
		}

		public static boolean storeCounter(String counterKey, long count) {
			logger.info("memcached store counter key: "+counterKey);
			boolean bool = false;
			try {
	        	THREAD_COUNT.incrementAndGet();
				bool = mcc.storeCounter(counterKey, count);
			} catch (Exception e) {
				// NOOP
			} finally {
	            THREAD_COUNT.decrementAndGet();
			}
			return bool;
		}

		public static long incr(String counterKey) {
			logger.info("memcached counter key: "+ counterKey);
			long count = -1;
			try {
	        	THREAD_COUNT.incrementAndGet();
				count = mcc.incr(counterKey);
			} catch (Exception e) {
				// NOOP
			} finally {
	            THREAD_COUNT.decrementAndGet();
			}
			return count;
		}

		public static Object get(String key) {
			logger.info("memcached key: "+key);
			if (isMCInitSuccess) {
			    long stime = System.currentTimeMillis();
			    Object obj = null;
				try {
		        	THREAD_COUNT.incrementAndGet();
					obj = mcc.get(key);
				} catch (Exception e) {
					// NOOP
				} finally {
		            THREAD_COUNT.decrementAndGet();
				}
			    /*Log007.logWapMemcached(key, stime, (int)(System.currentTimeMillis() - stime));*/
			    return obj;
			}
			else{
				logger.error("miscmemcached isMCInitSuccess false in get()");
				return cache.get(key);
			}
		}


	    public static Map<String,Object> getMulti(String[] keys) {
	        if (keys != null && keys.length > 0) {
	            if (isMCInitSuccess) {
	                long stime = System.currentTimeMillis();
	                Map<String,Object> map = null;
					try {
			        	THREAD_COUNT.incrementAndGet();
						map = mcc.getMulti(keys);
					} catch (Exception e) {
						// NOOP
					} finally {
			            THREAD_COUNT.decrementAndGet();
					}
	                /*Log007.logWapMemcached("Array:" + keys[0], stime, (int)(System.currentTimeMillis() - stime));*/
	                return map;
	            }
	            else {
	                Map<String,Object> map = new HashMap<String,Object>();
	                for (String key : keys) map.put(key, cache.get(key));
	                return map;
	            }
	        }
	        else return null;
	    }


	    public static void set(String key, Object value) {
	    	logger.info("memcached set key: "+key);
	    	if(value==null)
	    		return;
	    	else if (isMCInitSuccess) {
			    long stime = System.currentTimeMillis();
			    try {
		        	THREAD_COUNT.incrementAndGet();
					mcc.set(key, value);
				} catch (Exception e) {
					// NOOP
				} finally {
		            THREAD_COUNT.decrementAndGet();
				}
				/*Log007.logWapMemcached(key, stime, (int)(System.currentTimeMillis() - stime));  */
			}
			else {
				logger.error("miscmemcached isMCInitSuccess false in set()");
				cache.put(key, value);
			}
		}

	    public static void set(String key, Object value, Date expiration) {
	    	logger.info("memcached set timeout key: "+key);
	    	if(value == null)
	    		return;
	    	else if (isMCInitSuccess) {
			    long stime = System.currentTimeMillis();
			    try {
		        	THREAD_COUNT.incrementAndGet();
					mcc.set(key, value, expiration);
				} catch (Exception e) {
					// NOOP
				} finally {
		            THREAD_COUNT.decrementAndGet();
				}
				/*Log007.logWapMemcached(key, stime, (int)(System.currentTimeMillis() - stime));*/
			}
			else {

				// do we have existing timer task for this key?
				// if yes, cancel it before setting up a new one
				AxCacheTimerTask cacheTimer = (AxCacheTimerTask)timerTasks.remove(key);
				if (cacheTimer != null) cacheTimer.cancel();

				// now replace the cache value with new one
				// setup the new timer task for the same key
				cache.put(key, value);
				cacheTimer = new AxCacheTimerTask(key);
				timerTasks.put(key, cacheTimer);
				axCacheTimer.schedule(cacheTimer, expiration);
				logger.error("miscmemcached isMCInitSuccess false in set()");
			}
		}

		public static Object delete(String key) {
			logger.info("memcached delete key: "+key);
			if (isMCInitSuccess) {
				Object obj = null;
				try {
		        	THREAD_COUNT.incrementAndGet();
					obj = mcc.delete(key);
				} catch (Exception e) {
					// NOOP
				} finally {
		            THREAD_COUNT.decrementAndGet();
				}
				return obj;
			}
			else return cache.remove(key);
		}

		public static boolean getIsMCInitSuccess() {
			return isMCInitSuccess;
		}

		public static void init() {
			logger.info("Initialize memcached ...");

			//memcached_servers=memcached1.blueapple.mobi:1624,memcached1.blueapple.mobi:1624,memcached1.blueapple.mobi:1624
			//memcached_server_weights=3,3,3
			String mcServers = Configure.AX_MEMCACHED_SERVERS;
			logger.info("mcServers string from properties:" + mcServers);
			String mcServerWeights = Configure.AX_MEMCACHED_SERVERS_WEIGHTS;
			logger.info("mcServerWeights=" + mcServerWeights);

			// server list and weights
			String[] servers = mcServers.split(",");
			logger.info("MCServers=" + Arrays.toString(servers));
			Integer[] weights = null;
			if (mcServerWeights != null && mcServerWeights.trim().length() > 0) {
				String [] weightsString =  mcServerWeights.split(",");
				if (weightsString != null && weightsString.length > 0) {
					weights = new Integer[weightsString.length];
					for (int i=0; i<weightsString.length; i++) {
						try {
							weights[i] = new Integer(weightsString[i]);
						} catch (NumberFormatException e) {
							// defaults to 3 on bad entry
							weights[i] = 3;
						}
					}
				}
			}
			logger.info("weights=" + Arrays.toString(weights));
			// grab an instance of our connection pool
			SockIOPool pool = SockIOPool.getInstance("AXMEMCACHED");

			// set the servers and the weights
			pool.setServers( servers );
			if (weights != null) pool.setWeights( weights );

			// TODO: put all the following parameters in ba.properties file

	        int initialConnections  = 50;
	        int minSpareConnections = 5;
	        int maxSpareConnections = 250;
	        long maxIdleTime        = 1000 * 60 * 30;       // 30 minutes
	        long maxBusyTime        = 1000 * 60 * 5;        // 5 minutes
	        long maintThreadSleep   = 1000 * 5;                     // 5 seconds
	        int     socketTimeOut       = 1000 * 3;                 // 3 seconds to block on reads
	        int     socketConnectTO     = 1000 * 3;                 // 3 seconds to block on initial connections.  If 0, then will use blocking connect (default)
	        boolean failover        = true;                        // turn off auto-failover in event of server down
	        boolean nagleAlg        = false;                        // turn off Nagle's algorithm on all sockets in pool
	        boolean aliveCheck      = false;                        // disable health check of socket on checkout

	        pool.setInitConn( initialConnections );
	        pool.setMinConn( minSpareConnections );
	        pool.setMaxConn( maxSpareConnections );
	        pool.setMaxIdle( maxIdleTime );
	        pool.setMaxBusyTime( maxBusyTime );
	        pool.setMaintSleep( maintThreadSleep );
	        pool.setSocketTO( socketTimeOut );
			pool.setSocketConnectTO( socketConnectTO );
	        pool.setNagle( nagleAlg );
	        pool.setFailover(failover);
	        pool.setAliveCheck(aliveCheck);
	        pool.setHashingAlg( SockIOPool.NEW_COMPAT_HASH );

			// initialize the connection pool
			pool.initialize();

			// lets set some compression on for the client
			// compress anything larger than 64k
			mcc.setCompressEnable( true );
			mcc.setCompressThreshold( 64 * 1024 );

			// check initialization status
			//check();

			logger.error("Initialized memcached status: " + isMCInitSuccess);
		}
}

